package io.vertx.kue.service.impl

import io.vertx.kue.queue.*
import io.vertx.kue.queue.JobState
import io.vertx.kue.service.JobService
import io.vertx.kue.util.RedisHelper
import io.vertx.core.AsyncResult
import io.vertx.core.Future
import io.vertx.core.Handler
import io.vertx.core.Vertx
import io.vertx.core.json.JsonArray
import io.vertx.core.json.JsonObject
import io.vertx.redis.RedisClient

import java.util.ArrayList
import java.util.stream.Collectors

/**
 * Redis backend implementation of [JobService].
 *
 * @author Eric Zhao
 */
class JobServiceImpl @JvmOverloads constructor(private val vertx: Vertx, private val config: JsonObject = JsonObject()) : JobService {
    private val client: RedisClient

    init {
        this.client = RedisClient.create(vertx, RedisHelper.options(config))
        Job.setVertx(vertx, RedisHelper.client(vertx, config)) // init static vertx instance inner job
    }

    override fun getJob(id: Long, handler: Handler<AsyncResult<Job>>): JobService {
        val zid = RedisHelper.createFIFO(id)
        client.hgetall(RedisHelper.getKey("job:" + id)) { r ->
            if (r.succeeded()) {
                try {
                    if (!r.result().containsKey("id")) {
                        handler.handle(Future.succeededFuture())
                    } else {
                        val job = Job(r.result())
                        job.setId(id)
                        job.setZid(zid)
                        handler.handle(Future.succeededFuture(job))
                    }
                } catch (e: Exception) {
                    e.printStackTrace()
                    this.removeBadJob(id, "", null)
                    handler.handle(Future.failedFuture(e))
                }

            } else {
                this.removeBadJob(id, "", null)
                handler.handle(Future.failedFuture(r.cause()))
            }
        }
        return this
    }

    override fun removeJob(id: Long, handler: Handler<AsyncResult<Void>>): JobService {
        this.getJob(id, Handler{ r ->
            if (r.succeeded()) {
                if (r.result() != null) {
                    r.result().remove()
                            .setHandler(handler)
                } else {
                    handler.handle(Future.succeededFuture())
                }
            } else {
                handler.handle(Future.failedFuture(r.cause()))
            }
        })
        return this
    }

    override fun existsJob(id: Long, handler: Handler<AsyncResult<Boolean>>): JobService {
        client.exists(RedisHelper.getKey("job:" + id)) { r ->
            if (r.succeeded()) {
                if (r.result() == 0L)
                    handler.handle(Future.succeededFuture(false))
                else
                    handler.handle(Future.succeededFuture(true))
            } else {
                handler.handle(Future.failedFuture(r.cause()))
            }
        }
        return this
    }

    override fun getJobLog(id: Long, handler: Handler<AsyncResult<JsonArray>>): JobService {
        client.lrange(RedisHelper.getKey("job:$id:log"), 0, -1, handler)
        return this
    }

    override fun jobRangeByState(state: String, from: Long, to: Long, order: String, handler: Handler<AsyncResult<List<Job>>>): JobService {
        return rangeGeneral("jobs:" + state.toUpperCase(), from, to, order, handler)
    }

    override fun jobRangeByType(type: String, state: String, from: Long, to: Long, order: String, handler: Handler<AsyncResult<List<Job>>>): JobService {
        return rangeGeneral("jobs:" + type + ":" + state.toUpperCase(), from, to, order, handler)
    }

    override fun jobRange(from: Long, to: Long, order: String, handler: Handler<AsyncResult<List<Job>>>): JobService {
        return rangeGeneral("jobs", from, to, order, handler)
    }

    /**
     * Range job by from, to and order
     *
     * @param key     range type(key)
     * @param from    from
     * @param to      to
     * @param order   range order(asc, desc)
     * @param handler result handler
     */
    private fun rangeGeneral(key: String, from: Long, to: Long, order: String, handler: Handler<AsyncResult<List<Job>>>): JobService {
        if (to < from) {
            handler.handle(Future.failedFuture<List<Job>>("to can not be greater than from"))
            return this
        }
        client.zrange(RedisHelper.getKey(key), from, to) { r ->
            if (r.succeeded()) {
                if (r.result().size() == 0) { // maybe empty
                    handler.handle(Future.succeededFuture(ArrayList<Job>()))
                } else {
                    val list = r.result().list.stream()
                            .map { e -> RedisHelper.numStripFIFO(e as String) }
                            .collect(Collectors.toList<Long>()) as java.util.List<Long>
                    list.sort { a1: Long, a2: Long ->
                        if (order == "asc")
                            java.lang.Long.compare(a1, a2)
                        else
                            java.lang.Long.compare(a2, a1)
                    }
                    val max = Math.max(list[0], list[list.size - 1])
                    val jobList = ArrayList<Job>()
                    list.forEach { e ->
                        this.getJob(e, Handler{ jr ->
                            if (jr.succeeded()) {
                                if (jr.result() != null) {
                                    jobList.add(jr.result())
                                }
                                if (e >= max) {
                                    handler.handle(Future.succeededFuture<List<Job>>(jobList))
                                }
                            } else {
                                handler.handle(Future.failedFuture(jr.cause()))
                            }
                        })
                    }
                }
            } else {
                handler.handle(Future.failedFuture<List<Job>>(r.cause()))
            }
        }
        return this
    }

    /**
     * Remove bad job by id (absolutely)
     *
     * @param id      job id
     * @param handler result handler
     */
    private fun removeBadJob(id: Long, jobType: String, handler: Handler<AsyncResult<Void>>?): JobService {
        val zid = RedisHelper.createFIFO(id)
        client.transaction().multi(null)
                .del(RedisHelper.getKey("job:$id:log"), null)
                .del(RedisHelper.getKey("job:" + id), null)
                .zrem(RedisHelper.getKey("jobs:INACTIVE"), zid, null)
                .zrem(RedisHelper.getKey("jobs:ACTIVE"), zid, null)
                .zrem(RedisHelper.getKey("jobs:COMPLETE"), zid, null)
                .zrem(RedisHelper.getKey("jobs:FAILED"), zid, null)
                .zrem(RedisHelper.getKey("jobs:DELAYED"), zid, null)
                .zrem(RedisHelper.getKey("jobs"), zid, null)
                .zrem(RedisHelper.getKey("jobs:$jobType:INACTIVE"), zid, null)
                .zrem(RedisHelper.getKey("jobs:$jobType:ACTIVE"), zid, null)
                .zrem(RedisHelper.getKey("jobs:$jobType:COMPLETE"), zid, null)
                .zrem(RedisHelper.getKey("jobs:$jobType:FAILED"), zid, null)
                .zrem(RedisHelper.getKey("jobs:$jobType:DELAYED"), zid, null)
                .exec { r ->
                    if (handler != null) {
                        if (r.succeeded())
                            handler.handle(Future.succeededFuture())
                        else
                            handler.handle(Future.failedFuture(r.cause()))
                    }
                }

        // TODO: add search functionality

        return this
    }

    override fun cardByType(type: String, state: JobState, handler: Handler<AsyncResult<Long>>): JobService {
        client.zcard(RedisHelper.getKey("jobs:" + type + ":" + state.name), handler)
        return this
    }

    override fun card(state: JobState, handler: Handler<AsyncResult<Long>>): JobService {
        client.zcard(RedisHelper.getKey("jobs:" + state.name), handler)
        return this
    }

    override fun completeCount(type: String, handler: Handler<AsyncResult<Long>>): JobService {
        return if (type == null)
            this.card(JobState.COMPLETE, handler)
        else
            this.cardByType(type, JobState.COMPLETE, handler)
    }

    override fun failedCount(type: String, handler: Handler<AsyncResult<Long>>): JobService {
        return if (type == null)
            this.card(JobState.FAILED, handler)
        else
            this.cardByType(type, JobState.FAILED, handler)
    }

    override fun inactiveCount(type: String, handler: Handler<AsyncResult<Long>>): JobService {
        return if (type == null)
            this.card(JobState.INACTIVE, handler)
        else
            this.cardByType(type, JobState.INACTIVE, handler)
    }

    override fun activeCount(type: String, handler: Handler<AsyncResult<Long>>): JobService {
        return if (type == null)
            this.card(JobState.ACTIVE, handler)
        else
            this.cardByType(type, JobState.ACTIVE, handler)
    }

    override fun delayedCount(type: String, handler: Handler<AsyncResult<Long>>): JobService {
        return if (type == null)
            this.card(JobState.DELAYED, handler)
        else
            this.cardByType(type, JobState.DELAYED, handler)
    }

    override fun getAllTypes(handler: Handler<AsyncResult<List<String>>>): JobService {
        client.smembers(RedisHelper.getKey("job:types")) { r ->
            if (r.succeeded()) {
                handler.handle(Future.succeededFuture(ArrayList<Any>(r.result().list) as List<String>))
            } else {
                handler.handle(Future.failedFuture(r.cause()))
            }
        }
        return this
    }

    override fun getIdsByState(state: JobState, handler: Handler<AsyncResult<List<Long>>>): JobService {
        client.zrange(RedisHelper.getStateKey(state), 0, -1) { r ->
            if (r.succeeded()) {
                val list = r.result().stream()
                        .map { e -> RedisHelper.numStripFIFO(e as String) }
                        .collect(Collectors.toList<Long>())
                handler.handle(Future.succeededFuture(list))
            } else {
                handler.handle(Future.failedFuture(r.cause()))
            }
        }
        return this
    }

    override fun getWorkTime(handler: Handler<AsyncResult<Long>>): JobService {
        client.get(RedisHelper.getKey("stats:work-time")) { r ->
            if (r.succeeded()) {
                handler.handle(Future.succeededFuture(java.lang.Long.parseLong(if (r.result() == null) "0" else r.result())))
            } else {
                handler.handle(Future.failedFuture(r.cause()))
            }
        }
        return this
    }
}
